package flink_p2_sql

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{Table, TableEnvironment}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.types.Row

object sql01_sqlquery {


  /**
   * 使用sqlapi
   *  按照年龄统计人数
   * @param args
   */
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.scala._

    val dataStream: DataStream[(Long, String, Int)] = env.socketTextStream("node1", 8889)
      .map(data => {
        val arr: Array[String] = data.split(" ")
        (arr(0).toLong, arr(1), arr(2).toInt)
      })

    /**
     * 1. 使用sql api查询
     */
//    val userTable: Table = tableEnv.fromDataStream(dataStream, 'time, 'name, 'age)
//
//    val resTable: Table = tableEnv.sqlQuery(s"select age,count(*) from ${userTable} group by age")
//    resTable.toRetractStream[Row].print()


    /**
     * 2. 使用sql api查询
     */
    tableEnv.registerDataStream("user_table", dataStream, 'id,'name,'age)   //注册table 并指定表名

    val resTable2: Table = tableEnv.sqlQuery("select age,count(*) from user_table group by age")

    resTable2.toRetractStream[Row].print()




    tableEnv.execute("sql api.")
  }
}
